package com.uuuo.rxjava2;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.view.View;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.ExecutionException;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
    }

    public void log(String log) {
        log("", log);
    }

    public void log(String tag, String log) {
        System.err.println(tag + "--->" + log);
    }


    public void flowable(View view) throws ExecutionException, InterruptedException {

        Flowable flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                log("", "subscribe");
                e.onNext(1);
                e.onNext(2);
                e.onNext(1);
                e.onNext(2);
                e.onComplete();
            }
        }, BackpressureStrategy.ERROR);

        Subscriber subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                log("onSubscribe");
                s.request(5);
            }

            @Override
            public void onNext(Integer integer) {
                log("onNext", integer + "");
            }

            @Override
            public void onError(Throwable t) {
                log("onError", t.toString());
            }

            @Override
            public void onComplete() {
                log("", "onComplete");
            }
        };

        flowable.subscribe(subscriber);

    }

    public void request(View view) {

    }


    public ObservableTransformer<Integer, Integer> applySchedulers() {
        return new ObservableTransformer<Integer, Integer>() {
            @Override
            public ObservableSource<Integer> apply(Observable<Integer> upstream) {
                return upstream.filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        System.err.println("applySchedulers--->" + Thread.currentThread().getName());
                        return integer > 3;
                    }
                }).subscribeOn(Schedulers.io()).observeOn(Schedulers.newThread());
            }
        };
    }

    public void compose(View view) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                System.err.println("subscribe--->" + Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onNext(4);
                e.onNext(5);
                e.onComplete();
            }
        }).compose(applySchedulers()).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.err.println("accept--->" + Thread.currentThread().getName());
                System.out.println(integer);
            }
        });
    }
}
